package com.hnyfkj.jyindustry.common;

import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.UnsupportedEncodingException;

public class DiscontanctMQTT {

    private static String topic="zhangzhang";
    // 服务器连接参数一般写在配置文件中
    private static String host="tcp://127.0.0.1:1883";
    private static String userName="zhang";
    private static String password="zhang";
    private static String clientId="weiwei";
    private static MqttClient  client =null;
    /**
     * 创建客户端
     */
    public static boolean createClient(){
        // 创建客户端, MemoryPersistence 设置 client 的保存方式默认以内存保存
        // clientId 是客户端的唯一标识, 相同的 clientId 客户端连接时会导致已连接的客户端断开连接
        try {
            client = new MqttClient(host,clientId,new MemoryPersistence());
            // 设置回调函数
            client.setCallback(new MqttCallbackExtended(){
                /**
                 * todo 本文重点 在连接成功回调函数中 向 topic 主题发送客户端上线通知
                 * @param b
                 * @param s
                 */
                @Override
                public void connectComplete(boolean b, String s) {
                    System.out.println("[MQTT Service] 连接已完成...");
                    // 连接完成可以做一些初始化操作
                    MqttMessage mqttMessage = new MqttMessage();
                    mqttMessage.setQos(1);
                    // 设置消息保留
                    mqttMessage.setRetained(true);
                    try {
                        mqttMessage.setPayload("online".getBytes("UTF-8"));
                    } catch (UnsupportedEncodingException e) {
                        e.printStackTrace();
                    }
                    try {
                        client.publish(topic,mqttMessage);
                    } catch (MqttException e) {
                        e.printStackTrace();
                    }
                }

                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("[MQTT Service] 连接已断开...");
                    // 不设置断线自动连接时，可在这边自定义重连
                }


                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
                    System.out.println("[MQTT Service] 数据已到达...");
                    String content = new String(message.getPayload(),"UTF-8");
                    System.out.println("【MQTT】订阅主题:" + topic);
                    System.out.println("【MQTT】订阅质量:" + message.getQos());
                    System.out.println("【MQTT】订阅内容:" + content);
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("[MQTT Service] 数据传送成功...");
                }
            });
            MqttConnectOptions connectOptions = new MqttConnectOptions();
            // 设置是否关闭连接会话
            // 设置为 true 时,客户端每次都会以新的身份连接
            // 设置为 false 时,客户端再次连接会继续之前的会话，再次连接后无需再次订阅即可收到之前已订阅过的主题消息，包括断线后发布的消息。
            // 设置为 false 有个前置条件就是 clientId 不变
            connectOptions.setCleanSession(true);
            connectOptions.setUserName(userName);
            connectOptions.setPassword(password.toCharArray());
            // 断线后是否自动重连
            connectOptions.setAutomaticReconnect(false);
            try {
                // todo 本文重点 设置遗嘱消息,并保留
                connectOptions.setWill(topic,"offline".getBytes("UTF-8"),1,true);
            } catch (UnsupportedEncodingException e) {
                e.printStackTrace();
            }
            // 客户端重复连接不会导致重复订阅和发布
            try {
                client.connect(connectOptions);
            } catch (MqttException e) {
                System.out.println("客户端连接失败！"+ e.getMessage());
                return false;
            }
        } catch (MqttException e) {
            e.printStackTrace();
        }

        return true;
    }


    /**
     * 发布消息
     * @param topicSuffix 主题后缀
     * @param message 消息内容
     */
    public static void publishTopic(String topicSuffix,String message){
        if(null == client && !createClient()){
            return;
        }
        MqttMessage mqttMessage = new MqttMessage();
        mqttMessage.setQos(1);
        try {
            mqttMessage.setPayload(message.getBytes("UTF-8"));
        } catch (UnsupportedEncodingException e) {
            e.printStackTrace();
        }
        // 设置消息保留
        // 设置为 true 则 Broker 会保留对该主题发生的最后一条消息，每有客户端订阅时都会推送一次
        // 设置为 true 后推送的那条消息会被一直保留，需要取消只有在设置为 true 下发送一条空的 payload 消息
        mqttMessage.setRetained(false);
        try {
            // 主题前缀用于区分同一主题的发送方向，防止自己订阅自己发布的主题
            client.publish(topicSuffix,mqttMessage);
        } catch (MqttException e) {
            e.printStackTrace();
        }
    }




}
